package com.huan.inputformat;


import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FSDataInputStream;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.ByteWritable;
import org.apache.hadoop.io.BytesWritable;
import org.apache.hadoop.io.IOUtils;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.InputSplit;
import org.apache.hadoop.mapreduce.RecordReader;
import org.apache.hadoop.mapreduce.TaskAttemptContext;
import org.apache.hadoop.mapreduce.lib.input.FileSplit;

import java.io.IOException;

public class WholeRecordReader extends RecordReader<Text, BytesWritable> {

    //TODO 用来做文件分割
    FileSplit split;
    Configuration conf ;
    Text k = new Text();
    BytesWritable v = new BytesWritable();
    boolean isProgress = true;

    @Override
    public void initialize(InputSplit split, TaskAttemptContext context) throws IOException, InterruptedException {

        //TODO 初始化
        this.split= (FileSplit) split;
        conf = context.getConfiguration();

    }

    @Override
    public boolean nextKeyValue() throws IOException, InterruptedException {

        if (isProgress){
            byte[] buf = new byte[(int) split.getLength()];
            //TODO 核心业务处理
            //TODO 1，获取fs对象
            Path path = split.getPath();
            FileSystem fs = path.getFileSystem( conf );

            //TODO 2,获取输入流
            FSDataInputStream fis = fs.open( path );

            //TODO 将他们拷贝
            //TODO 长度是buf的长度
            IOUtils.readFully( fis,buf,0,buf.length );


            //TODO 封装value
            v.set(buf, 0, buf.length);


            //TODO 封装key
            k.set( path.toString() );

            isProgress = false;

            return true;
        }




        return false;
    }

    @Override
    public Text getCurrentKey() throws IOException, InterruptedException {
        //TODO 返回key
        return k;
    }

    @Override
    public BytesWritable getCurrentValue() throws IOException, InterruptedException {
        //TODO 返回value
        return v;
    }

    @Override
    public float getProgress() throws IOException, InterruptedException {
        return 0;
    }

    @Override
    public void close() throws IOException {

    }
}
